package com.hcc.Spark_FPG;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.fpm.AssociationRules;
import org.apache.spark.mllib.fpm.FPGrowth;
import org.apache.spark.mllib.fpm.FPGrowthModel;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class JavaSimpleFPGrowth {

	private static JavaSparkContext sc;

	public static void main(String[] args) throws SQLException {

		// 执行SQL,创建文件
		// getAll();

		SparkConf conf = new SparkConf().setAppName("FP-growth Example");
		sc = new JavaSparkContext(conf);

		// 待分析数据 TODO 目前是读取工程下的文件
		JavaRDD<String> data = sc.textFile("data/mllib/sample_fpgrowth.txt");

		JavaRDD<List<String>> transactions = data.map(new Function<String, List<String>>() {
			public List<String> call(String line) {
				String[] parts = line.split(",");
				return Arrays.asList(parts);
			}
		});

		// 设置频率(支持率)下限
		FPGrowth fpg = new FPGrowth().setMinSupport(0.5).setNumPartitions(3);
		FPGrowthModel<String> model = fpg.run(transactions);

		for (FPGrowth.FreqItemset<String> itemset : model.freqItemsets().toJavaRDD().collect()) {
			System.out.println("[" + itemset.javaItems() + "], " + itemset.freq()
					+ "	##########################################");
		}

		double minConfidence = 0;// 置信下限
		save(model, minConfidence);

		sc.stop();
	}

	private static Connection getConn() {
		String driver = "com.mysql.jdbc.Driver";
		String url = "jdbc:mysql://202.123.72.168:3306/jeeshop";
		String username = "root";
		String password = "Hitachi@123";
		Connection conn = null;
		try {
			Class.forName(driver); // classLoader,加载对应驱动
			conn = (Connection) DriverManager.getConnection(url, username, password);
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return conn;
	}

	private static ResultSet getAll() {
		Connection conn = getConn();
		String sql = "SELECT group_concat(DISTINCT productID) as productID FROM t_comment group by accountID";
		PreparedStatement pstmt;
		ResultSet rs = null;

		String porductID = "";
		PrintWriter pw = null;
		FileWriter fw = null;
		try {
			pstmt = (PreparedStatement) conn.prepareStatement(sql);
			rs = pstmt.executeQuery();
			String filename = "data/mllib/sample_fpgrowth.txt";
			File file = new File(filename);

			file.delete();// 删除文件
			while (rs.next()) {
				porductID = rs.getString("productID");
				if (file.exists()) {
					// 判断文件存在就写数据
					try {
						fw = new FileWriter(filename, true);
						if (rs.isLast()) {
							fw.write(porductID);
						} else {
							fw.write(porductID + "\n");
						}
						fw.flush();
					} catch (IOException e) {
						e.printStackTrace();
					}
				} else {
					// 判断文件不存在就new新文件,写数据
					try {
						file.createNewFile();
						pw = new PrintWriter(file);
						if (rs.isLast()) {
							pw.write(porductID);
						} else {
							pw.write(porductID + "\n");
						}
						pw.flush();
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			if (conn != null) {
				try {
					// 关闭数据库连接
					conn.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
			if (pw != null) {
				// 关闭IO流
				pw.close();
			}
			if (fw != null) {
				try {
					fw.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return rs;
	}

	// 关联分析结果存入到Mysql
	public static void save(FPGrowthModel<String> model, double minConfidence) throws SQLException {
		Connection con = getConn();
		String tableName = "fpGrowth";
		con.prepareStatement("DROP TABLE IF EXISTS " + tableName).execute();
		StringBuilder createQuery = new StringBuilder("CREATE TABLE IF NOT EXISTS " + tableName + "(");
		createQuery.append("id" + " varchar(255) NOT NULL,");
		createQuery.append("product1" + " varchar(255) default NULL,");
		createQuery.append("product2" + " varchar(255) default NULL,");
		createQuery.append("confidence" + " double default NULL,");
		createQuery.append("PRIMARY KEY  (id));");
		con.prepareStatement(createQuery.toString()).execute();
		Statement statement = con.createStatement();

		StringBuilder insertQuery = new StringBuilder();

		try {
			for (AssociationRules.Rule<String> rule : model.generateAssociationRules(minConfidence).toJavaRDD()
					.collect()) {
				System.out.println(rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
				if (insertQuery != null) {
					insertQuery.delete(0, insertQuery.length());
				}

				insertQuery.append("INSERT INTO " + tableName + " VALUES (");
				insertQuery.append("'" + UUID.randomUUID().toString().replaceAll("-", "") + "', ");
				insertQuery.append("'" + rule.javaAntecedent() + "', ");
				insertQuery.append("'" + rule.javaConsequent() + "', ");
				insertQuery.append("'" + rule.confidence() + "')");
				statement.executeUpdate(insertQuery.toString());
			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			con.close();
			System.out.println("执行完成###################");
		}
	}
}